//package com.example.demo;
//
//import com.alibaba.ververica.connectors.datahub.sink.DatahubSinkFunction;
//import com.aliyun.datahub.client.model.*;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//
//
//public class DemoTest {
//    public static void main(String[] args) throws Exception {
//        //创建Flink流处理执行环境
//        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
//        //设置并行度
//        environment.setParallelism(1);
//        //调用Flink自定义Source
//        DataStreamSource<Sick> source = environment.addSource(new DataSource());
//        //打印数据
//        source.print();
//        //DataHub连接配置。
//        DatahubSinkFunction<RecordEntry> sinkFunction = new DatahubSinkFunction<>(
//                 "***************************************",//EndPoint
//                "*************",//DataHub项目名称
//                "*************",//主题topic的名称
//                "**********",//accessID
//                "************"//accessKey
//                );
//        //序列化信息
//        DataStream<RecordEntry> operator = source.map(new MapFunction<Sick, RecordEntry>() {
//            @Override
//            public RecordEntry map(Sick sick) throws Exception {
//                RecordEntry entry = getRecordEntry(sick);
//                return entry;
//            }
//        });
//        operator.addSink(sinkFunction);
//        //启动程序
//        environment.execute();
//    }
//    //序列化方法
//    public static RecordEntry getRecordEntry(Sick sick) {
//        //注册Schema信息
//        RecordSchema recordSchema = new RecordSchema();
//        recordSchema.addField(new Field("name", FieldType.STRING));
//        recordSchema.addField(new Field("sex", FieldType.STRING));
//        recordSchema.addField(new Field("age", FieldType.BIGINT));
//        recordSchema.addField(new Field("area", FieldType.BIGINT));
//        recordSchema.addField(new Field("disease_status", FieldType.BIGINT));
//        recordSchema.addField(new Field("date", FieldType.STRING));
//        RecordEntry recordEntry = new RecordEntry();
//        TupleRecordData recordData = new TupleRecordData(recordSchema);
//        //将数据流里面的对象根据下标赋值
//        recordData.setField(0, sick.getName());
//        recordData.setField(1, sick.getSex());
//        recordData.setField(2, sick.getAge());
//        recordData.setField(3, sick.getArea());
//        recordData.setField(4, sick.getDisease_status());
//        recordData.setField(5, sick.getDate());
//        recordEntry.setRecordData(recordData);
//        return recordEntry;
//    }
//}
